package com.atguigu.watermark;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
 * @author gmd
 * @desc 断点式水位生成器，用来设置无序流中的断点水位线
 * @since 2024-11-30 03:25:42
 */
public class MyPunctuatedGenerator<T> implements WatermarkGenerator<T> {

    // 乱序等待时间
    private long delayTs;

    // 用来保存 当前为止 最大的事件时间
    private long maxTs;

    public MyPunctuatedGenerator(long delayTs) {
        this.delayTs = delayTs;
        this.maxTs = Long.MIN_VALUE + this.delayTs + 1;
    }

    /**
     * 每条数据来，都会调用一次：该方法用于处理每条数据的到来，提取最大事件时间，保存下来，并生成watermark。
     *
     * @param event           当前事件数据
     * @param eventTimestamp  当前事件的时间戳
     * @param output          水位线输出
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        maxTs = Math.max(maxTs, eventTimestamp);
        output.emitWatermark(new Watermark(maxTs - delayTs - 1));
        System.out.println("调用onEvent方法，获取目前为止的最大时间戳=" + maxTs+",watermark="+(maxTs - delayTs - 1));
    }

    /**
     * 不需要实现onPeriodicEmit方法，进行周期性调用。
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // nothing
    }

}
